package com.eyas.im.client.nio;

import com.eyas.im.client.IMClient;
import com.eyas.im.client.nio.handler.AbstractSocketClientHandler;
import com.eyas.im.client.nio.handler.ChannelCloseException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.net.InetSocketAddress;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

/**
 * 非线程安全
 */
public class NIOIMClient implements IMClient {
    private final static Logger logger = LoggerFactory.getLogger(NIOIMClient.class);
    private final static NioClientConfiguration DEFAULT_CLIENT_CONFIGURATION = new NioClientConfiguration();
    private SocketChannel socketChannel;
    private Selector selector;
    private volatile ClientStatus status = ClientStatus.READY;
    private NioClientConfiguration clientConfiguration;
    private CountDownLatch countDownLatch = new CountDownLatch(1); //用于shutdown时保证安全关闭时间
    private AbstractSocketClientHandler<?> socketClientHandler;
    private int nioBugCounter = 0; //连续触发nioBug可能性判定的次数
    private long lastAcceptMils = 0L; //最后一次accept发生的次数


    public NIOIMClient(AbstractSocketClientHandler<?> socketClientHandler) {
        this.socketClientHandler = socketClientHandler;
        this.clientConfiguration = DEFAULT_CLIENT_CONFIGURATION;
        this.socketClientHandler.observer(this);
    }

    public NIOIMClient(AbstractSocketClientHandler<?> socketClientHandler, NioClientConfiguration nioClientConfiguration) {
        this.socketClientHandler = socketClientHandler;
        this.clientConfiguration = nioClientConfiguration;
        this.socketClientHandler.observer(this);
    }

    /**
     * 建立连接
     *
     * @param hostname
     * @param port
     * @throws Exception
     */
    @Override
    public void connect(String hostname, int port) throws Exception {
        if (this.status != ClientStatus.READY)
            throw new IllegalStateException("当前Client实例已经启动过，请重新new实例");
        try {
            this.socketChannel = SocketChannel.open();
            this.socketChannel.configureBlocking(false);
            this.selector = Selector.open();
            this.countDownLatch = new CountDownLatch(1);
            this.socketChannel.connect(new InetSocketAddress(hostname, port));
            this.socketChannel.register(selector, SelectionKey.OP_CONNECT);
        } catch (IOException e) {
            if (this.selector != null)
                this.selector.close();
            if (this.socketChannel != null)
                this.socketChannel.close();
            throw e;
        }
        this.status = ClientStatus.STARTED;

        //启动处理器
        this.exeucte();
    }

    private void exeucte() {
        try {
            while (this.status == ClientStatus.STARTED) {
                int selectSize = this.selector.select(this.clientConfiguration.getSelectorBlockTime());
                if (this.status == ClientStatus.CLOSED)
                    break;
                if (selectSize > 0) {
                    this.processEvent();
                } else {
                    if (this.IsjavaNioBugOccur()) {
                        this.changeSelector();
                    }
                }
            }
        } catch (IOException | SelectorChangeException | NoSuchMethodException | IllegalAccessException | InvocationTargetException | InstantiationException e) {
            logger.error("IOReactor异常", e);
            countDownLatch.countDown();
            shutdwonGraceful();
        } catch (ChannelCloseException e) {
            logger.debug("server端主动断开");
            countDownLatch.countDown();
            shutdwonGraceful();
        }
        countDownLatch.countDown();
        logger.debug("exeucte执行结束");
    }

    /**
     * java NIO在linux epoll模型下，是否触发空轮询的bug
     *
     * @return
     */
    private boolean IsjavaNioBugOccur() {
        if (!this.clientConfiguration.isNioBugChecker())
            return false;
        if (System.currentTimeMillis() - this.lastAcceptMils <= this.clientConfiguration.getNioBugIntervalThreshold()) {
            this.nioBugCounter++;
        } else {
            this.nioBugCounter = 0;
        }
        return this.nioBugCounter >= this.clientConfiguration.getNioBugPossibleCount();
    }

    /**
     * 将现在的selector的所有注册信息，copy到新的selector中，并关闭旧的selector
     * 这个方法主要是当java nio bug出现时使用的
     */
    private void changeSelector() throws SelectorChangeException {
        logger.warn("发生java NIO的空轮询bug，更换selector");
        Selector oldSelecotr = this.selector;
        try {
            Selector newSelector = Selector.open();
            for (SelectionKey selectionKey : oldSelecotr.keys()) {
                SelectableChannel channel = selectionKey.channel();
                int intrestOps = selectionKey.interestOps();
                Object attachment = selectionKey.attachment();
                selectionKey.cancel();
                channel.register(newSelector, intrestOps, attachment);
            }
            this.selector = newSelector;
        } catch (IOException e) {
            throw new SelectorChangeException("selector更换失败", e);
        } finally {
            try {
                oldSelecotr.close();
            } catch (IOException e) {
                logger.warn("select更换时，oldSelector关闭失败，这可能造成内存泄露问题", e);
            }
        }
    }

    private void processEvent() throws IOException, NoSuchMethodException, IllegalAccessException, InvocationTargetException, InstantiationException, ChannelCloseException {
        Iterator<SelectionKey> selectionKeyIterator = this.selector.selectedKeys().iterator();
        while (selectionKeyIterator.hasNext()) {
            SelectionKey selectionKey = selectionKeyIterator.next();
            selectionKeyIterator.remove();
            switch (selectionKey.readyOps()) {
                case SelectionKey.OP_ACCEPT:
                    logger.warn("client channel获取到accept事件，不应该！！！！");
                    break;
                case SelectionKey.OP_CONNECT:
                    logger.debug("client获取到OP_CONNECT事件");
                    long startTimeMilns = System.currentTimeMillis();
                    /**
                     * 循环体内的内容理论上不会出现，出现的话需要分析一下
                     */
                    while (!this.socketChannel.finishConnect()) {
                        logger.debug("client端finishconnect失败");
                        if (System.currentTimeMillis() - startTimeMilns > 60 * 1000)
                            throw new IOException("建立连接失败");
                    }
                    this.socketClientHandler.attch(selectionKey, this.socketChannel);
                    this.socketClientHandler.afterConnected();
                    //连接成功
                    selectionKey.interestOps(SelectionKey.OP_READ);
                    break;
                case SelectionKey.OP_READ:
                    logger.debug("client获取到OP_READ事件");
                    try {
                        this.socketClientHandler.onDataRecieved(this.socketClientHandler.read());
                    } catch (IOException e) {
                        logger.warn("当前终端从[" + socketChannel.getRemoteAddress() + "]读消息时出现问题", e);
                        this.socketClientHandler.beforeClose();
                        throw e;
                    }
                    break;
                case SelectionKey.OP_WRITE:
                    logger.debug("client获取到OP_WRITE事件");
                    try {
                        this.socketClientHandler.write();
                    } catch (IOException e) {
                        logger.warn("当前终端向[" + socketChannel.getRemoteAddress() + "]写消息时出现问题", e);
                        this.socketClientHandler.beforeClose();
                        throw e;
                    }
                    break;
            }
        }
    }

    @Override
    public void shutdwonGraceful() {
        if (this.status != ClientStatus.STARTED)
            return;
        this.status = ClientStatus.CLOSED;
        this.selector.wakeup();
        try {
            this.countDownLatch.await(this.clientConfiguration.getShutdownWaitTime(), TimeUnit.MILLISECONDS);
        } catch (InterruptedException e) {
            logger.warn("安全关闭阻塞被打断");
        }
        try {
            this.selector.close();
        } catch (IOException e) {
            logger.error("关闭selector失败", e);
        }
        try {
            this.socketChannel.close();
        } catch (IOException e) {
            logger.error("关闭socketchannel失败", e);
        }
        logger.debug("NIO Client关闭");
    }
}
